
#include "stdafx.h"
#include "ConstDef.h"
#include "ReloadCode.h"
#include "ReloadApp.h"
#include "ReloadPacket.h"
#include "ReloadOverlay.h"
#include "ReloadUdpCallPacket.h"
#include "ReloadUdpCallMgr.h"
#include "Reload.h"

namespace reload
{
	ReloadUdpCallMgr::ReloadUdpCallMgr( boost::weak_ptr<ReloadOverlay> _pReloadOverlay ,std::size_t _workerPoolSize /*= 1*/ )
	:asyframe::ActorMgrBase(_workerPoolSize),pReloadOverlay_(_pReloadOverlay)
	{

	}

	ReloadUdpCallMgr::~ReloadUdpCallMgr()
	{

	}

	retT ReloadUdpCallMgr::start()
	{
		for(int i=0; i < ReloadUdpCall_WorkerPoolSize; i++)
		{
			boost::shared_ptr<asyframe::ActorBase> pActor(new ReloadUdpCallActor(i));
			R(makeActor(pActor));
		}
		R(asyframe::ActorMgrBase::start());
		RETSUCCESS;
	}

	////////////////////////////////////////////////////////////////////////////////////////

	ReloadUdpCallActor::ReloadUdpCallActor(int _duty)
		:duty_(_duty),pUdpHandler_()
	{

	}

	ReloadUdpCallActor::~ReloadUdpCallActor()
	{
		duty_ = -1;
		pUdpHandler_->reset();
	}

	retT ReloadUdpCallActor::init()
	{
		R(asyframe::ActorBase::init());
		pUdpHandler_.reset(new asyframe::UdpHandler());
		boost::shared_ptr<boost::asio::ip::udp::endpoint> pLocalEndpoint(new boost::asio::ip::udp::endpoint(
			boost::asio::ip::udp::v4(), reloadUdpCallPort[duty_]));
		pUdpHandler_->setLocalEndpoint(pLocalEndpoint);
		R(pUdpHandler_->attachHandlerOwner(shared_from_this()));
		RETSUCCESS;
	}

	retT ReloadUdpCallActor::start()
	{
		R(asyframe::ActorBase::start());
		R(readOneReloadUdpCallRequest());
		RETSUCCESS;
	}

	retT ReloadUdpCallActor::readOneReloadUdpCallRequest()
	{
		if(boost::shared_ptr<asyframe::WorkerBase> pWorker = getWorker())
		{
			boost::shared_ptr<asyframe::UdpHandlerContext> pUdpHandlerContext(new asyframe::UdpHandlerContext());
			boost::shared_ptr<asyframe::IoBuffer> pBuffer(new asyframe::IoBuffer(pWorker, maxReloadUdpCallPacketSize));
			pUdpHandlerContext->bytesToTransfer_ = maxReloadUdpCallPacketSize;
			pUdpHandlerContext->pRemoteEndpoint_.reset(new boost::asio::ip::udp::endpoint());
			R(pUdpHandler_->asyncReadFrom(pBuffer, pUdpHandlerContext->pRemoteEndpoint_, pUdpHandlerContext,
				boost::bind(&ReloadUdpCallActor::onReadReloadUdpCallRequest, boost::dynamic_pointer_cast<ReloadUdpCallActor>(shared_from_this()),
				pUdpHandlerContext, _1, _2, _3)));
		}
		else
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		RETSUCCESS;
	}

	retT ReloadUdpCallActor::onReadReloadUdpCallRequest( boost::shared_ptr<asyframe::UdpHandlerContext> _pUdpHandlerContext, std::size_t _bytesTransferred, const boost::system::error_code& e, std::size_t _timeoutCount )
	{
		R(IUdpAction::onReadFrom(_pUdpHandlerContext, _bytesTransferred, e, _timeoutCount));
		R(readOneReloadUdpCallRequest());
		if(!e)
		{
			if(_pUdpHandlerContext && _pUdpHandlerContext->pRemoteEndpoint_)
			{
				if(_pUdpHandlerContext->pBuffer_)
				{
					R(processReloadUdpCallRequest(_pUdpHandlerContext->pBuffer_, _pUdpHandlerContext->pRemoteEndpoint_));
				}
			}
		}
		else
		{
			LOG_INFO(boost::format( "errorNo %1% : %2% in %3% %4% %5%" ) % e.value() % e.message() % __FUNCTION__ % __FILE__ % __LINE__);
		}
		RETSUCCESS;
	}

	retT ReloadUdpCallActor::processReloadUdpCallRequest( boost::shared_ptr<asyframe::IoBuffer> _pBuffer, boost::shared_ptr<boost::asio::ip::udp::endpoint> _pRemoteEndpoint )
	{
		switch(duty_)
		{
		case ReloadUdpCall_Store:R(processStoreReq(_pBuffer, _pRemoteEndpoint));break;
		case ReloadUdpCall_Fetch:R(processFetchReq(_pBuffer, _pRemoteEndpoint));break;
		case ReloadUdpCall_AppAttach:R(processAppAttachReq(_pBuffer, _pRemoteEndpoint));break;
		default:break;
		}
		RETSUCCESS;
	}

	retT ReloadUdpCallActor::processStoreReq( boost::shared_ptr<asyframe::IoBuffer> _pBuffer, boost::shared_ptr<boost::asio::ip::udp::endpoint> _pRemoteEndpoint )
	{
		boost::shared_ptr<SendStoreReqUdpCallPacket> pSendStoreReqUdpCallPacket(new SendStoreReqUdpCallPacket());
		R(pSendStoreReqUdpCallPacket->readFromBuffer(_pBuffer));
		if(pSendStoreReqUdpCallPacket->pReloadPacket_ && pSendStoreReqUdpCallPacket->pReloadPacket_->msgContents_.messageBody_.elements_.size() > 0)
		{
			boost::shared_ptr<reload::StoreReq> pStoreReq = boost::dynamic_pointer_cast<reload::StoreReq>(pSendStoreReqUdpCallPacket->pReloadPacket_->msgContents_.messageBody_.elements_[0]);
			if(pStoreReq)
			{
				R(sendStoreReq(pSendStoreReqUdpCallPacket->overlayInstanceName_.toString(), pSendStoreReqUdpCallPacket->resourceName_.toString().c_str(),
					pStoreReq->kindData_, boost::bind(&ReloadUdpCallActor::onSendStoreReqUserInvoke,
					boost::dynamic_pointer_cast<ReloadUdpCallActor>(shared_from_this()), _pRemoteEndpoint, _1, _2)));
			}
		}
		RETSUCCESS;
	}

	retT ReloadUdpCallActor::processFetchReq( boost::shared_ptr<asyframe::IoBuffer> _pBuffer, boost::shared_ptr<boost::asio::ip::udp::endpoint> _pRemoteEndpoint )
	{
		boost::shared_ptr<SendFetchReqUdpCallPacket> pSendFetchReqUdpCallPacket(new SendFetchReqUdpCallPacket());
		R(pSendFetchReqUdpCallPacket->readFromBuffer(_pBuffer));
		if(pSendFetchReqUdpCallPacket->pReloadPacket_ && pSendFetchReqUdpCallPacket->pReloadPacket_->msgContents_.messageBody_.elements_.size() > 0)
		{
			boost::shared_ptr<reload::FetchReq> pFetchReq = boost::dynamic_pointer_cast<reload::FetchReq>(pSendFetchReqUdpCallPacket->pReloadPacket_->msgContents_.messageBody_.elements_[0]);
			if(pFetchReq)
			{
				R(sendFetchReq(pSendFetchReqUdpCallPacket->overlayInstanceName_.toString(), pSendFetchReqUdpCallPacket->resourceName_.toString().c_str(),
					pFetchReq->specifiers_, boost::bind(&ReloadUdpCallActor::onSendFetchReqUserInvoke,
					boost::dynamic_pointer_cast<ReloadUdpCallActor>(shared_from_this()), _pRemoteEndpoint, _1, _2)));
			}
		}
		RETSUCCESS;
	}

	retT ReloadUdpCallActor::processAppAttachReq( boost::shared_ptr<asyframe::IoBuffer> _pBuffer, boost::shared_ptr<boost::asio::ip::udp::endpoint> _pRemoteEndpoint )
	{
		boost::shared_ptr<ReloadUdpCallPacketBase> pReloadUdpCallPacketBase(new ReloadUdpCallPacketBase());
		R(pReloadUdpCallPacketBase->readFromBuffer(_pBuffer));
		R(sendAppAttachReq(pReloadUdpCallPacketBase->overlayInstanceName_.toString(), pReloadUdpCallPacketBase->resourceName_.toString().c_str(),
			boost::bind(&ReloadUdpCallActor::onSendAppAttachReqUserInvoke,
			boost::dynamic_pointer_cast<ReloadUdpCallActor>(shared_from_this()), _pRemoteEndpoint, _1, _2)));
		RETSUCCESS;
	}

	retT ReloadUdpCallActor::onSendStoreReqUserInvoke( boost::shared_ptr<boost::asio::ip::udp::endpoint> _pRemoteEndpoint, reload::TransactionStatus _transactionStatus, boost::shared_ptr<reload::ReloadNetworkDataContext> _pReloadNetworkDataContext )
	{
		dispatch(boost::bind(&ReloadUdpCallActor::onSendAppAttachReq,
			boost::dynamic_pointer_cast<ReloadUdpCallActor>(shared_from_this()), _pRemoteEndpoint, _transactionStatus, _pReloadNetworkDataContext));
		RETSUCCESS;
	}

	retT ReloadUdpCallActor::onSendFetchReqUserInvoke( boost::shared_ptr<boost::asio::ip::udp::endpoint> _pRemoteEndpoint, reload::TransactionStatus _transactionStatus, boost::shared_ptr<reload::ReloadNetworkDataContext> _pReloadNetworkDataContext )
	{
		dispatch(boost::bind(&ReloadUdpCallActor::onSendFetchReq,
			boost::dynamic_pointer_cast<ReloadUdpCallActor>(shared_from_this()), _pRemoteEndpoint, _transactionStatus, _pReloadNetworkDataContext));
		RETSUCCESS;
	}

	retT ReloadUdpCallActor::onSendAppAttachReqUserInvoke( boost::shared_ptr<boost::asio::ip::udp::endpoint> _pRemoteEndpoint, reload::TransactionStatus _transactionStatus, boost::shared_ptr<reload::ReloadNetworkDataContext> _pReloadNetworkDataContext )
	{
		dispatch(boost::bind(&ReloadUdpCallActor::onSendAppAttachReq,
			boost::dynamic_pointer_cast<ReloadUdpCallActor>(shared_from_this()), _pRemoteEndpoint, _transactionStatus, _pReloadNetworkDataContext));
		RETSUCCESS;
	}

	retT ReloadUdpCallActor::onSendStoreReq( boost::shared_ptr<boost::asio::ip::udp::endpoint> _pRemoteEndpoint, reload::TransactionStatus _transactionStatus, boost::shared_ptr<reload::ReloadNetworkDataContext> _pReloadNetworkDataContext )
	{
		if(_pRemoteEndpoint)
		{
			ReloadUdpCallResultBase reloadUdpCallResultBase;
			if(_transactionStatus.status_ == reload::TransactionStatus_Success)
			{
				if(_pReloadNetworkDataContext && _pReloadNetworkDataContext->pReloadPacket_ && _pReloadNetworkDataContext->pReloadPacket_->msgContents_.messageCode_ != reload::MessageCode_Error)
				{
					reloadUdpCallResultBase.result_ =0; 
				}
				else
				{
					if(_pReloadNetworkDataContext && _pReloadNetworkDataContext->pReloadPacket_ && _pReloadNetworkDataContext->pReloadPacket_->msgContents_.messageCode_ == reload::MessageCode_Error)
					{
						reloadUdpCallResultBase.result_ = _pReloadNetworkDataContext->pReloadPacket_->msgContents_.errorResponse_.errorCode_;
					}
					else
					{
						reloadUdpCallResultBase.result_ = -1;
					}
				}
			}
			else
			{
				reloadUdpCallResultBase.result_ = -1;
			}
			if(boost::shared_ptr<asyframe::WorkerBase> pWorker = getWorker())
			{
				boost::shared_ptr<asyframe::UdpHandlerContext> pUdpHandlerContext(new asyframe::UdpHandlerContext());
				pUdpHandlerContext->bytesToTransfer_ = reloadUdpCallResultBase.size();
				boost::shared_ptr<asyframe::IoBuffer> pBuffer(new asyframe::IoBuffer(pWorker, pUdpHandlerContext->bytesToTransfer_));
				R(reloadUdpCallResultBase.writeToBuffer(pBuffer));
				R(pUdpHandler_->asyncWriteTo(pBuffer, _pRemoteEndpoint, pUdpHandlerContext));
			}
			else
				RET(ERROR_COMMON_SHARED_PTR_NULL);
		}
		RETSUCCESS;
	}

	retT ReloadUdpCallActor::onSendFetchReq( boost::shared_ptr<boost::asio::ip::udp::endpoint> _pRemoteEndpoint, reload::TransactionStatus _transactionStatus, boost::shared_ptr<reload::ReloadNetworkDataContext> _pReloadNetworkDataContext )
	{
		if(_pRemoteEndpoint)
		{
			SendFetchReqUdpCallResult sendFetchReqUdpCallResult;
			if(_transactionStatus.status_ == reload::TransactionStatus_Success)
			{
				if(_pReloadNetworkDataContext && _pReloadNetworkDataContext->pReloadPacket_ && _pReloadNetworkDataContext->pReloadPacket_->msgContents_.messageCode_ != reload::MessageCode_Error)
				{
					sendFetchReqUdpCallResult.result_ = 0; 
					sendFetchReqUdpCallResult.pReloadPacket_ = _pReloadNetworkDataContext->pReloadPacket_;
				}
				else
				{
					if(_pReloadNetworkDataContext && _pReloadNetworkDataContext->pReloadPacket_ && _pReloadNetworkDataContext->pReloadPacket_->msgContents_.messageCode_ == reload::MessageCode_Error)
					{
						sendFetchReqUdpCallResult.result_ = _pReloadNetworkDataContext->pReloadPacket_->msgContents_.errorResponse_.errorCode_;
					}
					else
					{
						sendFetchReqUdpCallResult.result_ = -1;
					}
				}
			}
			else
			{
				sendFetchReqUdpCallResult.result_ = -1;
			}
			if(boost::shared_ptr<asyframe::WorkerBase> pWorker = getWorker())
			{
				boost::shared_ptr<asyframe::UdpHandlerContext> pUdpHandlerContext(new asyframe::UdpHandlerContext());
				pUdpHandlerContext->bytesToTransfer_ = sendFetchReqUdpCallResult.size();
				boost::shared_ptr<asyframe::IoBuffer> pBuffer(new asyframe::IoBuffer(pWorker, pUdpHandlerContext->bytesToTransfer_));
				R(sendFetchReqUdpCallResult.writeToBuffer(pBuffer));
				R(pUdpHandler_->asyncWriteTo(pBuffer, _pRemoteEndpoint, pUdpHandlerContext));
			}
			else
				RET(ERROR_COMMON_SHARED_PTR_NULL);
		}
		RETSUCCESS;
	}

	retT ReloadUdpCallActor::onSendAppAttachReq( boost::shared_ptr<boost::asio::ip::udp::endpoint> _pRemoteEndpoint, reload::TransactionStatus _transactionStatus, boost::shared_ptr<reload::ReloadNetworkDataContext> _pReloadNetworkDataContext )
	{
		if(_pRemoteEndpoint)
		{
			SendAppAttachReqUdpCallResult sendAppAttachReqUdpCallResult;
			if(_transactionStatus.status_ == reload::TransactionStatus_Success)
			{
				if(_pReloadNetworkDataContext && _pReloadNetworkDataContext->pReloadPacket_ && _pReloadNetworkDataContext->pReloadPacket_->msgContents_.messageCode_ != reload::MessageCode_Error)
				{
					sendAppAttachReqUdpCallResult.result_ =0; 
					if(_pReloadNetworkDataContext && _pReloadNetworkDataContext->pReloadPacket_ && _pReloadNetworkDataContext->pReloadPacket_->msgContents_.messageBody_.elements_.size() > 0)
					{
						if(boost::shared_ptr<reload::AppAttachAns> pAppAttachAns = boost::dynamic_pointer_cast<reload::AppAttachAns>(_pReloadNetworkDataContext->pReloadPacket_->msgContents_.messageBody_.elements_[0]))
						{
							for(std::size_t i = 0; i < pAppAttachAns->candidates_.array_.size(); ++i)
							{
								sendAppAttachReqUdpCallResult.strIP_.setValue(pAppAttachAns->candidates_.array_[i].addrPort_.addr_.to_string());
							}
						}
					}
				}
				else
				{
					if(_pReloadNetworkDataContext && _pReloadNetworkDataContext->pReloadPacket_ && _pReloadNetworkDataContext->pReloadPacket_->msgContents_.messageCode_ == reload::MessageCode_Error)
					{
						sendAppAttachReqUdpCallResult.result_ = _pReloadNetworkDataContext->pReloadPacket_->msgContents_.errorResponse_.errorCode_;
					}
					else
					{
						sendAppAttachReqUdpCallResult.result_ = -1;
					}
				}
			}
			else
			{
				sendAppAttachReqUdpCallResult.result_ = -1;
			}
			if(boost::shared_ptr<asyframe::WorkerBase> pWorker = getWorker())
			{
				boost::shared_ptr<asyframe::UdpHandlerContext> pUdpHandlerContext(new asyframe::UdpHandlerContext());
				pUdpHandlerContext->bytesToTransfer_ = sendAppAttachReqUdpCallResult.size();
				boost::shared_ptr<asyframe::IoBuffer> pBuffer(new asyframe::IoBuffer(pWorker, pUdpHandlerContext->bytesToTransfer_));
				R(sendAppAttachReqUdpCallResult.writeToBuffer(pBuffer));
				R(pUdpHandler_->asyncWriteTo(pBuffer, _pRemoteEndpoint, pUdpHandlerContext));
			}
			else
				RET(ERROR_COMMON_SHARED_PTR_NULL);
		}
		RETSUCCESS;
	}
}
